package com.ht.api.flink;

import java.util.Objects;
import java.util.Properties;

//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MysqlCDC {

//    @Data
//    public static class Config {
//        // MySQL配置
//        private String hostname = "127.0.0.1";
//        private String username = "root";
//        private String password = "aixi_2028";
//        private int port = 3306;
//        private String database = "test";
//        private String table = "users";
//    }
//
//    public static void main(String[] args) throws Exception {
//        Config config = new Config();
//        
//        // 1. 设置环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        
//        // 检查点的作用：
//		//        故障恢复：当程序崩溃时，可以从最近的检查点恢复
//		//        数据一致性：确保数据不会丢失或重复处理
//		//        状态管理：保存程序的处理状态
//		//        在 MySQL CDC 的场景中，检查点可以：
//		//        记录已经读取的 binlog 位置
//		//        在程序重启时从上次保存的位置继续读取（在 Flink CDC 中，检查点的位置信息默认是保存在内存中的，如果要持久化检查点数据，需要配置检查点的存储路径。）
//		//        避免数据丢失或重复消费
//        
//        env.enableCheckpointing(3000); // 启用检查点机制，每 3000 毫秒（3秒）触发一次检查点
//        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 设置精确一次语义
//        checkpointConfig.setMinPauseBetweenCheckpoints(500); // 最小检查点间隔500毫秒
//        checkpointConfig.setCheckpointTimeout(60000); // 检查点超时时间1分钟
//        checkpointConfig.setMaxConcurrentCheckpoints(1); // 同一时间只允许进行一个检查点
//        
//        // 设置检查点存储路径
//		//        checkpointConfig.setCheckpointStorage("file:///path/to/checkpoint/dir");
//		//        // 或者
//		//        checkpointConfig.setCheckpointStorage(new Path("/path/to/checkpoint/dir"));
//        checkpointConfig.setCheckpointStorage("file:///D:/Flink/checkpoint"); // 检查点的保存位置，注意必须以file:/// 开头（三个斜杠）
//        
//        // 可能重复消费的情况：
//		//        - 程序非正常退出（如强制关闭）时，最后一个检查点之后的数据可能会重复消费
//		//        网络故障导致检查点保存失败
//		//        检查点存储目录被意外删除或损坏
//		// 可能丢失数据的情况：
//		//        MySQL binlog 被清理（超过 expire_logs_days）
//		//        MySQL 主从切换
//		//        检查点间隔太长，程序崩溃时丢失间隔期间的数据
//        
//        // 设置重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
//        
//        // 2. MySQL Source
//        Properties jdbcProperties = new Properties();
//        jdbcProperties.setProperty("connectTimeout", "30000");
//        jdbcProperties.setProperty("socketTimeout", "30000");
//        jdbcProperties.setProperty("useSSL", "false");
//
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//            .hostname(config.getHostname())
//            .port(config.getPort())
//            .databaseList(config.getDatabase())
//            .tableList(config.getDatabase() + "." + config.getTable())
//            .username(config.getUsername())
//            .password(config.getPassword())
//            .deserializer(new JsonDebeziumDeserializationSchema())
//            
//            // 选项1：1.先进行全量数据的快照读取（抓取现有数据）2.然后再进行增量数据的读取（监听新变更）
//            .startupOptions(StartupOptions.initial()) 
//            // 选项2：只读取最新数据（不读取历史数据）
//            //.startupOptions(StartupOptions.latest())
//            // 选项3：从特定时间点开始读取
//            //.startupOptions(StartupOptions.timestamp(1577808000000L))  // 时间戳，毫秒
//            // 选项4：从特定位置开始读取
//            //.startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 120L))
//            
//            // MySQL 的主从复制需要每个复制客户端有唯一的 server ID  注意：server ID 必须是唯一的，不能与 MySQL 主从复制中的其他 server ID 冲突。
//            .serverId("5400-5440")
//            
//            .jdbcProperties(jdbcProperties)
//            .build();
//            
//        // 3. 数据处理和打印
//        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC")
//        .map(value -> {
//        	log.info("原始数据：{}", value);
//        	String formatMsg = formatDebeziumLog(value);
//        	log.info("格式后：{}", formatMsg);
//        	return formatMsg;
//        })
//        .filter(Objects::nonNull);
//        //.print();
//           
//        // 4. 执行
//        env.execute("MySQL CDC Demo");
//    }
//    
//    /**
//     * 格式化 Debezium 日志
//     * @param log 原始 JSON 日志字符串
//     * @return 格式化后的日志字符串
//     */
//    public static String formatDebeziumLog(String log) {
//        // 解析原始日志为 JSONObject
//        JSONObject rawJson = JSONUtil.parseObj(log);
//
//        // 获取操作类型
//        String op = rawJson.getByPath("op", String.class);
//        if (StrUtil.isBlank(op)) {
//            return null;
//        }
//
//        String operationType;
//        switch (op) {
//	        case "r":
//	        	operationType = "read";
//	        	break;
//            case "c":
//                operationType = "insert";
//                break;
//            case "u":
//                operationType = "update";
//                break;
//            case "d":
//                operationType = "delete";
//                break;
//            default:
//                operationType = "none";
//        }
//
//        // 获取表和数据库信息
//        String table = rawJson.getByPath("source.table", String.class);
//        String database = rawJson.getByPath("source.db", String.class);
//
//        // 获取变化前后的数据
//        JSONObject before = rawJson.getByPath("before", JSONObject.class);
//        JSONObject after = rawJson.getByPath("after", JSONObject.class);
//
//        // 获取时间戳
//        Long timestamp = rawJson.getByPath("ts_ms", Long.class);
//
//        // 构造简洁的格式化输出
//        JSONObject formattedLog = new JSONObject();
//        formattedLog.set("type", operationType);
//        formattedLog.set("database", database);
//        formattedLog.set("table", table);
//        formattedLog.set("time", timestamp);
//        formattedLog.set("before", before);
//        formattedLog.set("after", after);
//
//        return formattedLog.toStringPretty();
//    }
}